package com.chb.flink.source

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import scala.util.Random


object CustomerSource {
    def main(args: Array[String]): Unit = {
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        import org.apache.flink.streaming.api.scala._

        // 设置checkpoint
        env.enableCheckpointing(1000)

        // 通过addSource
        val stream: DataStream[StationLog] = env.addSource(new MyCustomerSource)

        stream.print()
        env.execute()
    }
}


/**
 * 当然也可以自定义数据源，有两种方式实现：
 *  通过实现 SourceFunction 接口来自定义无并行度（也就是并行度只能为 1）的 Source。
 *  通过实现 ParallelSourceFunction 接口或者继承 RichParallelSourceFunction 来自定义有并行度的数据源。
 * *
 * 写一个实现SourceFunction接口
 */
class MyCustomerSource extends SourceFunction[StationLog] {
    // 定义一个是否中止流的标记
    var flag = true;

    override def run(sourceContext: SourceFunction.SourceContext[StationLog]): Unit = {

        var random = new Random();
        var types = Array("fail", "busy", "barring", "success")
        while (flag) { // 如果没有中止， 就一直发送数据
            1.to(2).map(i => {
                var callOut = ("183005%05d".format(random.nextInt(100000)))
                var callIn = ("188005%05d".format(random.nextInt(100000)))
                var a = Array(-1, 1)
                val time = System.currentTimeMillis() + random.nextInt(10) * 1000 * a(random.nextInt(2))
                val df = new SimpleDateFormat("HH:mm:ss")
                val stationLog = new StationLog("sid" + random.nextInt(10), callOut, callIn, types(random.nextInt(4)), time, random.nextInt(1000))
                println("MyCustomerSource: " + df.format(new Date(time)) + " " + stationLog)
                stationLog
            }).foreach(sourceContext.collect(_)) // 发送数据
            Thread.sleep(2000)
        }

    }

    // 中止数据流
    override def cancel(): Unit = {
        flag = false
    }
}